package edu.gmu.middleware;

import edu.gmu.messages.Client;
import edu.gmu.messages.EndPoint;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

/**
 * The CallbackStub is similar to RemoteStub except it's use is in the
 * "opposite direction" and is inherently asymchronous - thus we need
 * a callback/thread.
 * 
 * @author  James H. Pope
 * @version $Revision:$ $Date:$
 */
public class CallbackStub //implements SinkListener
{
    private Sink sink                = null;
    private volatile boolean stopped = false;
    
    private Marshaler marshaler      = new MarshalerImpl();
    private Unmarshaler unmarshaler  = new UnmarshalerImpl();
    
    private ByteBuffer in            = null;
    private ByteBuffer out           = null;
    
    private MessageRegistry registry = new MessageRegistry();
    
    private Client client            = null;
    private List callbackListeners   = new ArrayList();
    
    private EndPoint endpoint        = null;
    
    /**
     * Creates a new RemoteStub specified values.
     * @param endpoint
     */
    public CallbackStub( EndPoint endpoint ) throws IOException
    {
        this.sink = SinkFactory.makeTCPSink(endpoint.address, endpoint.port);
        this.sink.connect();
        
        this.client = new Client();
        this.client.id   = endpoint.port;
        this.client.name = endpoint.address;
        
        /*
         * Register RouterMessage bindings
         */
        //this.register(CONNECT,     Connect.class);
        //this.register(DISCONNECT,  Disconnect.class);
        
        /*
         *  NOTE: BIG ALLOCATIONS HERE
         */
        this.in  = ByteBuffer.allocate( RemoteStub.DEFAULT_BUFFER_SIZE );
        // Write has own buffer so cannopt fail due to lack of buffer
        this.out = ByteBuffer.allocate( RemoteStub.DEFAULT_BUFFER_SIZE );
        
        this.start();
    }
    
    public EndPoint getEndPoint()
    {
        return this.endpoint;
    }
    
    public void start()
    {
        /*
         * Starts reading
         */
        Runnable task = new Runnable()
        {
            public void run()
            {
                while( stopped == false )
                {
                    try
                    {
                        receive();
                    }
                    catch( IOException ioe )
                    {
                        //ioe.printStackTrace();
                        //System.out.println( "CallbackStub - Callback disconnected." );
                        return;
                    }
                    
                }
            }
        };
        Thread t = new Thread(task);
        t.setName("CallbackThread");
        t.start();
    }
    
    /**
     * Registers binding between given id and class
     * @param messageId
     * @param class to bind to messageId
     * @returns the newly registered MessageBinding
     * @throws IllegalArgumentException if id is already bound
     */
    public MessageBinding register( int topic, Class boundClass, Callback callback )
    {
        MessageBinding messageBinding = this.registry.register(topic, boundClass);
        CallbackBinding callbackBinding = new CallbackBinding( messageBinding, callback );
        this.callbackListeners.add( callbackBinding );
        return messageBinding;
    }
    
    /**
     * Unregisters ???
     * @param topic
     * @return
     */
    public boolean unregister( int topic )
    {
        return this.registry.unregister(topic);
    }
    
    /**
     * Performs the following actions - all on the caller's thread.
     * 1. Marshals requestMessage to byteBuffer out
     * 2. Writes out to sink
     * 3. Reads in from sink to get response (assumes remots side responds)
     * 4. Unmarshals response into responseMessage
     * 5. return responseMessage
     * @param topic
     * @param message
     * @throws IOException if the unmashaling/write process was not successful
     * @throws IllegalStateException if no MessageBinding for response
     */
//    public Message execute( int requestTopic, Message requestMessage ) throws IOException
//    {
//        this.send(requestTopic, requestMessage);
//        /*
//         * Now switch over and read from the sink for the response
//         * This is what makes this synchronous
//         */
//        Message responseMessage = receive();
//        return responseMessage;
//    }
    
    /**
     * Performs the following actions - all on the caller's thread.
     * 1. Marshals requestMessage to byteBuffer out
     * 2. Writes out to sink
     * 3. Reads in from sink to get response (assumes remots side responds)
     * 4. Unmarshals response into responseMessage
     * 5. return responseMessage
     * @param topic
     * @param message
     * @throws IOException if the mashaling/write process was not successful
     */
    public void send( int requestTopic, Message requestMessage ) throws IOException
    {
        System.out.print("Sending...");
        Header requestHeader = new Header();
        requestHeader.setTopic(requestTopic);
        this.marshaler.marshal(requestHeader, requestMessage, out);
        
        //print(out,12);
        this.sink.write(out);
        System.out.print("Done Sending.");
    }
    
    /**
     * Performs the following actions - all on the caller's thread.
     * 1. Reads in from sink to get response (assumes remots side responds)
     * 2. Unmarshals response into responseMessage
     * 3. return responseMessage
     * @throws IOException if the read/unmashaling process was not successful
     * @throws IllegalStateException if no MessageBinding for response
     */
    public Message receive() throws IOException
    {
        /*
         * Now switch over and read from the sink for the response
         * This is what makes this synchronous
         */
        Header responseHeader = this.sink.read(in);

        //System.out.println("Client received..." + responseHeader.toString() );
        //print( inBuffer, 12 );

        int responseTopic = responseHeader.getTopic();
        MessageBinding messageBinding = registry.getMessageBinding(responseTopic);
        if( messageBinding == null )
        {
            String s = "Cannot locate MessageBinding for "+responseTopic+
                    " have you called register(topic, class)?";
            throw new IllegalStateException(s);
        }
        
        // This will takes a while
        Message responseMessage = unmarshaler.unmarshal(responseHeader,
                                                in,
                                                messageBinding);
        
        this.fireCallbacks(responseHeader, responseMessage);
        
        return responseMessage;
    }
    
    
    /**
     * Determines if underlying sink isConnected.
     * @return true if connected, false otherwise
     */
    public boolean isConnected()
    {
        return this.sink.isConnected();
    }
    
    
    /**
     * Called from source object when the sink is disconnected.
     * @param event
     */
    public void sinkDisconnected( SinkDisconnectedEvent event )
    {
        // Regardless of disconnect type - go ahead and shutdown
        // Another option is if exception, try to reconnect
        //--------------------------------------------------------------------//
        // Called on Sink readerThread - last statement it executes
        //--------------------------------------------------------------------//
        this.stop();
        //--------------------------------------------------------------------//
    }
    
    
    
    /**
     * Stops the RemoteStub from sending/receiving. Also results in
     * the sink being stopped.
     */
    public void stop( )
    {
        this.stopped = true;
        this.sink.disconnect();
    }
    
    
    /**
     * Notify all listeners that have registered interest for notification
     * on this event.
     * @param header
     * @param message
     */
    public void fireCallbacks( Header header, Message message )
    {
        //Do this in order to only create Event once
        //MessageEvent event = null;

        //Notify listeners last to first, notifying those that are interested
        for(int i = 0; i < this.callbackListeners.size(); i++)
        {
            // Get listener
            CallbackBinding listener = ((CallbackBinding)this.callbackListeners.get(i));
            // Call listener
            if( header.getTopic() == listener.getMessageBinding().getTopic() )
            {
                Callback callback = listener.getCallback();
                callback.execute(message);
            }
        }
    }

}
